//the leafnode implementation
package leafnode

import (
	"errors"
	"hash/crc32"
	"io"
	"net"
	"os"
	"strconv"
	"sync/atomic"
	"time"

	"rockstore"
	"util/config"
	"util/log"
	"util/zkwrapper"
)

type Server struct {
	ip      string
	port    string
	zk      string
	datadir string
	metadir string
	monitor string

	weight    string //availability weight
	rg        int    //replica group
	id        int    //own id
	isLeader  bool
	syncWrite bool

	zkw *zkwrapper.ZkWrapper

	fcp   *FollowersConnPool
	store *rockstore.Store
	stats *Stats

	state int32

	//the loggers
	errLog  *log.Logger
	infoLog *log.Logger
	workLog *log.Logger
}

func (s *Server) initLog() error {
	return nil
}

func (s *Server) checkConfig(cfg *config.Config) error {
	s.ip = cfg.GetString("ip")
	s.port = cfg.GetString("port")
	s.datadir = cfg.GetString("datadir")
	s.metadir = cfg.GetString("metadir")

	if s.ip == "" || s.port == "" || s.datadir == "" || s.metadir == "" {
		return errors.New("Bad Config File")
	}

	s.zk = cfg.GetString("zk")
	s.monitor = cfg.GetString("mon")
	idStr := cfg.GetString("id")

	if s.zk == "" || idStr == "" {
		return errors.New("Bad Config File")
	}

	s.id, _ = strconv.Atoi(idStr)
	//check validity of server id
	if s.id <= 0 || s.id > MaxServerId {
		return errors.New("BadServerId")
	}

	s.rg = GrpId(s.id)
	s.isLeader = IsLeader(s.id)

	s.syncWrite = ("true" == cfg.GetString("sync"))

	return nil
}

//atomic load/set
func (s *Server) setState(state int32) {
	atomic.StoreInt32(&s.state, state)
}
func (s *Server) getState() int32 {
	return atomic.LoadInt32(&s.state)
}

// server main operations - New, Start, and Shutdown
func NewServer() *Server {
	return new(Server)
}

func (s *Server) Start(cfg *config.Config) error {
	//firstly check the config values
	e := s.checkConfig(cfg)
	if e != nil {
		return e
	}
	//log
	e = s.initLog()
	if e != nil {
		return e
	}

	//storage engine

	//follower conn pool
	s.fcp = NewFollowersConnPool()

	//server stats
	s.stats = NewStats(s.id)

	//the initial state
	s.setState(ReplGroupSplit)

	//leader is responsible for building replication relationship with followers
	if s.isLeader {
		go func() {
			s.serveReplGroupChange()
		}()
	} else {
		go func() {
			for {
				s.ReportStats()
				s.checkLogRotation()
				s.checkDiskStatus()
				s.checkZkConnHealth()
				<-time.After(NormalWakeUpPeriod * time.Minute)
			}
		}()
	}

	//start to serve
	return s.listenAndServe()
}

func (s *Server) Shutdown() {
	s.setState(ShuttingDown)
}

func (s *Server) listenAndServe() error {
	//firstly, listen on the port
	l, errListen := net.Listen("tcp", ":"+s.port)
	if errListen != nil {
		s.errLog.Println("Error: failed to listen ", errListen)
		return errListen
	}

	//then serve the connections
	for {
		conn, errAccept := l.Accept()
		if errAccept != nil {
			s.errLog.Println("Error: failed to accept ", errAccept)
			return errAccept
		}
		if s.getState() == ShuttingDown {
			break
		}

		//now handle the connection
		go func() {
			s.serveConn(conn)
		}()
	}

	//close the socket
	l.Close()
	return nil
}

//serve a single connection
func (s *Server) serveConn(rc net.Conn) {
	s.stats.AddConnection()
	remoteAddr := rc.RemoteAddr().String()
	s.workLog.Println("Info: accepted a connection from " + remoteAddr)

	c, ok := rc.(*net.TCPConn)
	if !ok {
		rc.Close()
		s.errLog.Println("Error: tcp conn convert failed")
	}

	c.SetNoDelay(true)

	for {
		//in case the server was shut down..
		if s.getState() == ShuttingDown {
			break
		}

		req := NewPacket()
		e := req.ReadFromConn(c)
		if e != nil {
			if e == io.EOF {
				s.workLog.Println("Info: the conn was closed by the peer")
			} else {
				s.workLog.Println("Error: failed to read from the conn, ", e)
			}

			break
		}

		var reply *Packet

		if req.Opcode == Read {
			reply = s.handleRead(req)
		} else {
			if s.getState() != ReplGroupReady {
				s.workLog.Println("Warn: Unready Repl Group")
				break
			}

			reply = s.handleWrite(req)
		}

		errWrite := reply.WriteToConn(c)
		if errWrite != nil {
			s.workLog.Println("Error: have to break the conn, ", errWrite)
			break
		}
	}

	//close the connection
	c.Close()
	s.stats.RemoveConnection()
	s.workLog.Println("Info: closed the connection from " + remoteAddr)
}
